
package kvsnode;

import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.LinkedTransferQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TransferQueue;

import util.Hasher;
import util.HasherImpl;
import util.RouteTracker;
import util.RouteTrackerImpl;

/**
 * This abstract class is to reduce the redundancy of the KVSNodes.
 * 
 * @author  
 * @version 5/4/2013
 */
public abstract class AbstractKVSNode implements KVSNode
{
  /**
   * A number of thread in each queue to process the messages.
   */
  public static final int THREAD_IN_EACH_QUEUE = 1;
  
  /**
   * A map of multiple blocking queues to implement message channels (port id, blocking queue).
   */
  private final Map<String, ExecutorService> my_queues =
      new HashMap<String, ExecutorService>();

  /**
   * A map to keep track the sent message with relevant Future object which was
   * sent to the clients.
   */
  private final Map<String, Future<Object>> my_sent_message_table =
      new HashMap<String, Future<Object>>();

  /**
   * This is a media which connects blocking queues to one thread.
   * 
   */
  private ExecutorService my_transfer_queue;

  /**
   * To hash a string input to get the key for the string to find the
   * appropriate node.
   */
  private final Hasher my_hasher;

  /**
   * To find the quickest next node to send the string to the appropriate node.
   */
  private final RouteTracker my_route_tracker;
  
  /**
   * A default environment setting.
   * 
   * - related instruction in the instruction document -
   * 
   * This means that you should not design or implement your system in ways that
   * will unnecessarily constrain you later. Examples of such unnecessary
   * constraints would be hard-coding the locations of data files, IP addresses,
   * numbers of nodes in your system, etc.Use the command line or configuration
   * files
   */
  private final Properties my_properties;

  /**
   * Total number of nodes in the ring. This information is to reduce the calculation to get
   * the total number of nodes from the value m.
   */
  private final int my_total_node_number;
  
  /**
   * A node id for the node.
   */
  private int my_node_id;

  /**
   * A actual node id for the node.
   */
  private int my_actual_node_id;

  /**
   * Constructor for the AbstractKVSNode.
   * 
   * @param the_properties a property file for the node.
   */
  public AbstractKVSNode(final Properties the_properties)
  {
    my_properties = the_properties;
    final int m = Integer.valueOf(the_properties.getProperty("m"));
    my_total_node_number = (int) Math.pow(2, m);
    my_hasher = new HasherImpl(m);
    my_route_tracker = new RouteTrackerImpl(m);
    initializeQueues(m);
  }
  
  /**
   * Initialize the blocking queues.
   * 
   * @param the_m the value of m (2^m is the number of nodes in the ring).
   */
  private void initializeQueues(final int the_m)
  {
    for (int i = 0; i < the_m; i++) 
    {
      final BlockingQueue<Runnable> queue = 
          new LinkedBlockingQueue<Runnable>(QUEUE_LENGTH_FOR_CHANNELS);
      // each thread pool has a thread which grab a message in each channel and
      // put it into the TransferQueue, and a blocking queue.
      my_queues.put(String.valueOf(i), 
                    new ThreadPoolExecutor(THREAD_IN_EACH_QUEUE, THREAD_IN_EACH_QUEUE, 
                                           ALIVE_SEC_OF_IDLE_THREAD, TimeUnit.SECONDS, queue));
    }
    final TransferQueue<Runnable> t_queue = new LinkedTransferQueue<Runnable>();
    my_transfer_queue =
        new ThreadPoolExecutor(THREAD_IN_EACH_QUEUE, THREAD_IN_EACH_QUEUE,
                               ALIVE_SEC_OF_IDLE_THREAD, TimeUnit.SECONDS, t_queue);
  }
  
  /**
   * {@inheritDoc}
   */
  @Override
  public Future<Object> sendQuery(final String the_message)
  {
    return null;
  }
  
  /**
   * {@inheritDoc}
   */
  @Override
  public void sendMessage(final int the_source_node_id, 
      final int the_dest_port_id, final Object the_message)
  {
    
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public Integer getActualNodeForThisNode()
  {
    return getActualNodeId();
  }

  /**
   * This method processes the message sent from a client or another node.
   * 
   * @param the_message a requested message.
   * @return an object as the result.
   */
  public Future<Object> processMessage(final Object the_message)
  {
    final int hash_value = my_hasher.getKey("hey");
    my_route_tracker.getNextHop(hash_value, getNodeId());
    return null;
  }

  /**
   * This method is to calculate which is the next node in the ring. This method
   * is necessary to make it return the next node id when it gets to the 2^m -1
   * node. This method returns 0 when it gets to the (2^m - 1)th node. 
   * 
   * @param the_node_id current node id.
   * @return the next node id.
   */
  public int getNextNodeInRing(final int the_node_id) 
  {
    int result;
    if (the_node_id == my_total_node_number - 1)
    {
      result = 0;
    }
    else
    {
      result = the_node_id + 1;
    }
    return result;
  }
  
  /**
   * A getter for Properties.
   * 
   * @return a properties object.
   */
  public Properties getProperties()
  {
    return my_properties;
  }

  /**
   * A getter for node id.
   * 
   * @return a node id.
   */
  public int getNodeId()
  {
    return my_node_id;
  }

  /**
   * A setter for node id.
   * 
   * @param the_node_id a node id.
   */
  public void setNodeId(final int the_node_id)
  {
    my_node_id = the_node_id;
  }

  /**
   * A getter for actual node id.
   * 
   * @return an actual node id.
   */
  public int getActualNodeId()
  {
    return my_actual_node_id;
  }

  /**
   * A setter for actual node id.
   * 
   * @param the_actual_node_id a actual node id.
   */
  public void setActualNodeId(final int the_actual_node_id)
  {
    my_actual_node_id = the_actual_node_id;
  }

  /**
   * Returns a number of total number of the nodes in the ring.
   * 
   * @return the number of nodes in the ring.
   */
  public int getTotalNodeNumber()
  {
    return my_total_node_number;
  }
  
}
